package com.learn.kafka.config;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.IntegerDeserializer;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.util.Properties;

/**
 * Kafka 消费者配置父类，封装通用配置
 *
 * @author: MI
 * @email: 448341911@qq.com
 * @createTime: 2025/8/7 21:09
 * @updateUser: MI
 * @updateTime: 2025/8/7 21:09
 * @updateRemark: 修改内容
 * @version: 1.0
 */
public class ConsumerProperties implements KafkaConfiguration {

    // 保存 Kafka 配置的 Properties 对象
    protected Properties props;

    // 构造方法：初始化默认配置
    public ConsumerProperties() {
        props = new Properties();
        // 初始化通用配置（可根据需求修改默认值）
        initDefaultConfig();
    }

    // 初始化默认配置
    private void initDefaultConfig() {
        // Kafka 集群地址,以逗号分割（默认值，子类可覆盖）
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.110.17:9092");
        // 消费组 ID（默认值，子类可覆盖）
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "mate-test-group");
        // 是否自动提交偏移量，只有commit之后才更新消费组的 offset
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");
        // 消费者自动提交间隔（毫秒）
        props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000");
        // 偏移量重置策略（最早位置） earliest | latest | none
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        // Key 反序列化器
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, IntegerDeserializer.class.getName());
        // Value 反序列化器
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
    }

    /**
     * 提供获取配置的方法，供子类使用
     *
     * @return Properties
     */
    @Override
    public Properties getProps() {
        return props;
    }

    /**
     * 可选：提供修改配置的方法（灵活扩展）
     *
     * @param key   配置键（如 ProducerConfig.BOOTSTRAP_SERVERS_CONFIG）
     * @param value 配置值
     */
    @Override
    public void setProperty(String key, String value) {
        props.put(key, value);
    }

    // 新增：创建并返回KafkaConsumer实例（消除重复代码）
    protected KafkaConsumer<Integer, String> createConsumer() {
        return new KafkaConsumer<>(this.getProps());
    }
}
